package com.edu.dsai.utils;
import com.edu.dsai.enums.SSEMsgType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
@Slf4j
public class SSEServer {
    /**
     * 使用map对象，关联用户id和sse的服务链接
     */
    private static Map<String, SseEmitter> sseClientsMap = new ConcurrentHashMap<>();
    /**
     * 统计总在线人数
     */
    private static AtomicInteger onlineCounts = new AtomicInteger(0);
    //region sse连接
    public static SseEmitter connect(String userId) {
        // 设置超时时间，0代表永不过期；默认30秒，超时未完成任务则会抛出异常
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 注册SSE的回调方法
        sseEmitter.onCompletion(completionCallback(userId));
        sseEmitter.onError(errorCallback(userId));
        sseEmitter.onTimeout(timeoutCallback(userId));
        sseClientsMap.put(userId, sseEmitter);
        log.info("当前创建新的SSE连接，用户id为{}", userId);
        // 总在线人数累加
        onlineCounts.getAndIncrement();
        return sseEmitter;
    }
    /**
     * SSE连接完成后的回调方法(关闭连接的时候调用)
     */
    private static Runnable completionCallback(String userId) {
        return () -> {
            log.info("SSE连接完成并结束，用户id为：{}", userId);
            removeConnection(userId);
        };
    }
    /**
     * SSE连接超时后的回调方法(连接超时的时候调用)
     */
    private static Runnable timeoutCallback(String userId) {
        return () -> {
            log.info("SSE连接超时，用户id为：{}", userId);
            removeConnection(userId);
        };
    }
    /**
     * SSE连接错误时的回调方法(连接错误的时候调用)
     */
    private static Consumer<Throwable> errorCallback(String userId) {
        return (Throwable) -> {
            log.info("SSE连接错误，用户id为：{}", userId);
            removeConnection(userId);
        };
    }
    /**
     * 删除连接
     */
    private static void removeConnection(String userId) {
        sseClientsMap.remove(userId);
        log.info("SSE连接被移除，移除的用户id为：{}", userId);
        // 累减累加人数
        onlineCounts.getAndDecrement();
    }
    //endregion
    //region 发送消息
    /**
     * 发送单条消息
     */
    public static void sendMessage(String userId, String message, SSEMsgType msgType) {
        if (CollectionUtils.isEmpty(sseClientsMap)) {
            return;
        }
        if (sseClientsMap.containsKey(userId)) {
            SseEmitter sseEmitter = sseClientsMap.get(userId);
            // 发送消息
            sendEmitterMessage(sseEmitter, userId, message, msgType);
        }
    }
    /**
     * 批量发送消息给所有人
     */
    public static void sendMessageToAllUsers(String message) {
        if (CollectionUtils.isEmpty(sseClientsMap)) {
            return;
        }
        sseClientsMap.forEach((userId, sseEmitter) -> {
            sendEmitterMessage(sseEmitter, userId, message, SSEMsgType.MESSAGE);
        });
    }
    /**
     * 使用SseEmitter推送消息
     */
    public static void sendEmitterMessage(SseEmitter sseEmitter, String userId, String message, SSEMsgType msgType) {
        try {
            SseEmitter.SseEventBuilder data = SseEmitter.event().id(userId).name(msgType.getType()).data(message);
            sseEmitter.send(data);
        } catch (IOException e) {
            log.error("用户【{}】的消息推送发生异常！", userId);
            removeConnection(userId);
        }
    }
    /**
     * ADD事件流式输出
     */
    public static void sendMessageAdd(String userId, String message, SSEMsgType msgType) {
        if (CollectionUtils.isEmpty(sseClientsMap)) {
            return;
        }
        if (sseClientsMap.containsKey(userId)) {
            SseEmitter sseEmitter = sseClientsMap.get(userId);
            // 发送消息
            sendEmitterMessage(sseEmitter, userId, message, msgType);
        }
    }
    /**
     * 关闭指定用户的连接
     */
    public static void stopServer(String userId) {
        if (CollectionUtils.isEmpty(sseClientsMap)) {
            return;
        }
        SseEmitter sseEmitter = sseClientsMap.get(userId);
        if (sseEmitter != null) {
            sseEmitter.complete();
            removeConnection(userId);
            log.info("连接关闭成功，被关闭的用户为{}", userId);
        } else {
            log.warn("当前连接无须关闭，请不要重复操作");
        }
    }
    //endregion
    //region 统计在线人数
    /**
     * 获得当前总连接在线人数
     */
    public static int getOnlineCounts() {
        return onlineCounts.intValue();
    }
    //endregion
}